package job.relevance.hostrank;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import mapred.base.TextMapper;

import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;

public class RLV_HostRank_Mapper extends TextMapper {

    private static final String HOSTRANK_TAG = "pHostRank=";

    private static Map<String, List<String>> map = null;

    @Override
    public void map(Text key, Text value) throws IOException {
        String k = key.toString();
        if (map.containsKey(k)) {
            String v = value.toString();
            int begin = v.indexOf(HOSTRANK_TAG) + HOSTRANK_TAG.length();
            int end = (v + "\t").indexOf("\t", begin);
            String hr = v.substring(begin, end);
            for (String q : map.get(k)) {
                this.output(new Text(q), new Text(hr));
            }
        }
    }

    @Override
    public void setUp() {
        map = new HashMap<String, List<String>>();

        try {
            String list = this.getConfig("global");

            BufferedReader br = new BufferedReader(new InputStreamReader(this
                    .getFileSystem().open(new Path(list))));
            for (String line = br.readLine(); line != null; line = br
                    .readLine()) {
                if (!line.trim().isEmpty()) {
                    int pos = line.indexOf("\t");
                    String k = line.substring(0, pos);
                    String v = line.substring(pos + 1);
                    if (!map.containsKey(k)) {
                        map.put(k, new ArrayList<String>());
                    }
                    map.get(k).add(v);
                }
            }
            br.close();
        } catch (IOException e) {
            e.printStackTrace();
            System.exit(255);
        }

        try {
            String list = this.getConfig("text");

            BufferedReader br = new BufferedReader(new InputStreamReader(this
                    .getFileSystem().open(new Path(list))));
            for (String line = br.readLine(); line != null; line = br
                    .readLine()) {
                if (!line.trim().isEmpty()) {
                    int pos = line.indexOf("\t");
                    String k = line.substring(0, pos);
                    String v = line.substring(pos + 1);
                    if (!map.containsKey(k)) {
                        map.put(k, new ArrayList<String>());
                    }
                    map.get(k).add(v);
                }
            }
            br.close();
        } catch (IOException e) {
            e.printStackTrace();
            System.exit(254);
        }
    }

    @Override
    public void tearDown() throws IOException {
    }

}
